package com.qf.bootproducer;

import com.qf.bootproducer.consts.RabbitMQConst;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitOperations;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.amqp.RabbitTemplateConfigurer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnSingleCandidate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.data.redis.core.StringRedisTemplate;

@SpringBootApplication
@Slf4j
public class BootProducerApplication {

    @Autowired
    private StringRedisTemplate redisTemplate;

    public static void main(String[] args) {
        SpringApplication.run(BootProducerApplication.class, args);
    }

    @Bean
    public RabbitTemplate rabbitTemplate(RabbitTemplateConfigurer configurer, ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        configurer.configure(rabbitTemplate, connectionFactory);

        //设置确认回调，保障消息能够正常发送到服务器的交换机（一定要记得开启确认支持）
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /*
                CorrelationData correlationData: 封住交换机、路由键、消息内容的对象
                ack: 消息是否正常到达交换机
                cause：消息失败原因
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                if (ack){
                    log.info("消息正常到达了交换机....");
                }else {
                    log.info("消息未到达了交换机....");
                    //重新发送 （考虑重试次数，重试次数达到3次或者5次，将消息保存数据库的重试表，利用定时任务进行重试）
                    ReturnedMessage returned = correlationData.getReturned();
                    String ex = returned.getExchange();
                    String rk = returned.getRoutingKey();
                    Message msg = returned.getMessage();
                    rabbitTemplate.send(ex,rk,msg);
                }
            }
        });

        //设置返回回调，保障消息能够从交换机正常到达队列（一定要记得开启返回回调支持）
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            //只有当路由队列失败时，会触发该回调
            @Override
            public void returnedMessage(ReturnedMessage returned) {
                Message msg = returned.getMessage();

                String key = RabbitMQConst.QUE_RETRY_PREFIX + msg.getMessageProperties().getMessageId();
                Long retryCount = redisTemplate.opsForValue().increment(key);

                if (retryCount != null && retryCount <= 3) {
                    log.info("消息到达队列失败，进行重发....");
                    //重试（考虑重试次数，重试次数达到3次或者5次，将消息保存数据库的重试表，利用定时任务进行重试）
                    String ex = returned.getExchange();
                    String rk = returned.getRoutingKey();
//                String rk = "A.abc";  //为了防止路由键错误，导致无限重发，这里将模拟失败的路由键修改为正确的
                    rabbitTemplate.send(ex, rk, msg);
                }else{
                    //将消息保存数据库的重试表，利用定时任务进行重试 (交换机、路由键、消息内容、消息id、时间戳，重试次数)
                    log.info("将消息写入数据库的重试表.....");
                }
            }
        });

        return rabbitTemplate;
    }

}
